package com.learn.sb.kafka.producer;


import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.UUID;

@Component
@EnableScheduling
@Slf4j
public class KafkaProducer {

	@Autowired
	private KafkaTemplate kafkaTemplate;

	@Scheduled(cron="0/5 * * * * ?")
	public void send() {
		String msg = UUID.randomUUID().toString();
		ListenableFuture future = kafkaTemplate.send("test-topic", msg);
		future.addCallback(new ListenableFutureCallback() {
			@Override
			public void onFailure(Throwable ex) {
				log.info("send fail :" + msg, ex);
			}

			@Override
			public void onSuccess(@Nullable Object result) {
				log.info("send success ==> " + msg, result);
			}
		});
	}
}
